🎖️GitЯра🎖️
Commit dda971b5ce81ed5dd9549dbaba1d8548bbfa4126
Parents : 4e30fb5
Author : James Rich <2199651+jamesarich@users.noreply.github.com>
Signature : Signature validation error
Date : 2026-08-11T19:10:30-07:00
Committer : GitHub <noreply@github.com>
Date : 2026-08-12T02:10:30Z
fix(messaging): time out orphaned "Sending..." messages into a retryable failure (#6630)
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Changes
8 files changed, 235 insertions(+), 2 deletions(-)
Diff
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt
index 709199b5a1..2cdcb9d721 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/MeshConnectionManagerImpl.kt
@@ -495,6 +495,7 @@ class MeshConnectionManagerImpl(
}
override fun onRadioConfigLoaded() {
+ packetHandler.rearmSendAckTimeouts()
scope.handledLaunch {
val queuedPackets = packetRepository.getQueuedPackets()
queuedPackets.forEach { packet ->
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/PacketHandlerImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/PacketHandlerImpl.kt
index 3a32a5bd6f..017f6a4a9c 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/PacketHandlerImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/manager/PacketHandlerImpl.kt
@@ -48,8 +48,11 @@ import org.meshtastic.core.repository.RadioInterfaceService
import org.meshtastic.proto.FromRadio
import org.meshtastic.proto.MeshPacket
import org.meshtastic.proto.QueueStatus
+import org.meshtastic.proto.Routing
import org.meshtastic.proto.ToRadio
+import kotlin.time.Duration
import kotlin.time.Duration.Companion.milliseconds
+import kotlin.time.Duration.Companion.minutes
import kotlin.time.Duration.Companion.seconds
import kotlin.uuid.Uuid
@@ -66,6 +69,19 @@ class PacketHandlerImpl(
companion object {
private val TIMEOUT = 5.seconds
+ /**
+ * Grace period after which a sent packet still [MessageStatus.ENROUTE] is stamped [Routing.Error.TIMEOUT]
+ * (retryable) instead of showing as sending forever. Generous — well past the radio's retransmit window — and
+ * matches iOS's sendAckTimeout so both apps time out alike.
+ */
+ internal val SEND_ACK_TIMEOUT = 5.minutes
+
+ /**
+ * Minimum re-arm delay on reconnect: the firmware's phone-queue backlog may still deliver the missing ACK/NAK
+ * just after the config handshake, so give it a moment before stamping a timeout.
+ */
+ internal val REARM_GRACE = 30.seconds
+
/**
* Firmware-internal `ErrorCode` (MeshTypes.h `ERRNO_SHOULD_RELEASE`) leaked into `QueueStatus.res`: "no error,
* but the packet should still be released". Firmware 2.8+ returns it for self-addressed packets, which are
@@ -89,6 +105,9 @@ class PacketHandlerImpl(
private val queueResponse = mutableMapOf<Int, CompletableDeferred<Boolean>>()
private val routingResponse = mutableMapOf<Int, CompletableDeferred<Boolean>>()
+ private val timeoutMutex = Mutex()
+ private val sendAckTimeoutJobs = mutableMapOf<Int, Job>()
+
override fun sendToRadio(p: ToRadio) {
Logger.d { "Sending to radio ${p.toPIIString()}" }
val b = p.encode()
@@ -258,12 +277,47 @@ class PacketHandlerImpl(
private fun changeStatus(packetId: Int, m: MessageStatus) = scope.handledLaunch {
if (packetId != 0) {
getDataPacketById(packetId)?.let { p ->
- if (p.status == m) return@handledLaunch
- packetRepository.value.updateMessageStatus(p, m)
+ if (p.status != m) {
+ packetRepository.value.updateMessageStatus(p, m)
+ }
+ if (m == MessageStatus.ENROUTE) {
+ scheduleSendAckTimeout(packetId)
+ }
+ }
+ }
+ }
+
+ override fun rearmSendAckTimeouts() {
+ scope.handledLaunch {
+ packetRepository.value.getEnroutePackets().forEach { p ->
+ val remaining = p.time + SEND_ACK_TIMEOUT.inWholeMilliseconds - nowMillis
+ scheduleSendAckTimeout(p.id, remaining.milliseconds.coerceAtLeast(REARM_GRACE))
}
}
}
+ /**
+ * A send whose routing ACK/NAK never reaches the app (typically because it was disconnected when the radio's
+ * response arrived) would stay ENROUTE — "Sending…" — forever. Stamp it as a retryable timeout instead; a late ACK
+ * still upgrades it via handleAckNak.
+ *
+ * One timer per packet: re-arming supersedes the pending one, so repeated reconnects cannot pile up timers for the
+ * same send. Timers deliberately survive a disconnect — the ack genuinely never arrived, and the resulting state is
+ * retryable — so a user who never reconnects still sees the send resolve.
+ */
+ private suspend fun scheduleSendAckTimeout(packetId: Int, delayFor: Duration = SEND_ACK_TIMEOUT) {
+ timeoutMutex.withLock {
+ sendAckTimeoutJobs.remove(packetId)?.cancel()
+ sendAckTimeoutJobs.values.removeAll { it.isCompleted }
+ sendAckTimeoutJobs[packetId] =
+ scope.handledLaunch {
+ delay(delayFor)
+ // Conditional in the DAO transaction: an ACK/NAK landing while this timer waited must win.
+ packetRepository.value.timeOutEnroutePacket(packetId, Routing.Error.TIMEOUT.value)
+ }
+ }
+ }
+
private suspend fun getDataPacketById(packetId: Int): DataPacket? = withTimeoutOrNull(1.seconds) {
var dataPacket: DataPacket? = null
while (dataPacket == null) {
diff --git a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt
index f08de3ea3c..9bb9c4affe 100644
--- a/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt
+++ b/core/data/src/commonMain/kotlin/org/meshtastic/core/data/repository/PacketRepositoryImpl.kt
@@ -113,6 +113,15 @@ class PacketRepositoryImpl(private val dbManager: DatabaseProvider, private val
dbManager.currentDb.value.packetDao().getAllDataPackets().filter { it.status == MessageStatus.QUEUED }
}
+ override suspend fun getEnroutePackets(): List<DataPacket> = withContext(dispatchers.io) {
+ dbManager.currentDb.value.packetDao().getAllDataPackets().filter { it.status == MessageStatus.ENROUTE }
+ }
+
+ // A null from withDb means no database was available, so nothing was timed out.
+ override suspend fun timeOutEnroutePacket(packetId: Int, routingError: Int): Boolean = withContext(dispatchers.io) {
+ dbManager.withDb { it.packetDao().timeOutEnroutePacket(packetId, routingError) } ?: false
+ }
+
suspend fun insertRoomPacket(packet: RoomPacket) {
withContext(dispatchers.io + NonCancellable) { dbManager.withDb { it.packetDao().insert(packet) } }
}
diff --git a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt b/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt
index bd97138817..41afc9443e 100644
--- a/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt
+++ b/core/data/src/commonTest/kotlin/org/meshtastic/core/data/manager/PacketHandlerImplTest.kt
@@ -20,9 +20,11 @@ import dev.mokkery.MockMode
import dev.mokkery.answering.returns
import dev.mokkery.answering.throws
import dev.mokkery.every
+import dev.mokkery.everySuspend
import dev.mokkery.matcher.any
import dev.mokkery.mock
import dev.mokkery.verify
+import dev.mokkery.verify.VerifyMode.Companion.exactly
import dev.mokkery.verifySuspend
import io.kotest.property.Arb
import io.kotest.property.arbitrary.int
@@ -33,7 +35,10 @@ import kotlinx.coroutines.test.StandardTestDispatcher
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.runTest
import org.meshtastic.core.common.di.asServiceScope
+import org.meshtastic.core.common.util.nowMillis
import org.meshtastic.core.model.ConnectionState
+import org.meshtastic.core.model.DataPacket
+import org.meshtastic.core.model.MessageStatus
import org.meshtastic.core.repository.MeshLogRepository
import org.meshtastic.core.repository.PacketRepository
import org.meshtastic.core.repository.RadioInterfaceService
@@ -42,6 +47,7 @@ import org.meshtastic.proto.Data
import org.meshtastic.proto.MeshPacket
import org.meshtastic.proto.PortNum
import org.meshtastic.proto.QueueStatus
+import org.meshtastic.proto.Routing
import org.meshtastic.proto.ToRadio
import kotlin.test.BeforeTest
import kotlin.test.Test
@@ -49,6 +55,7 @@ import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertNotNull
import kotlin.test.assertTrue
+import kotlin.time.Duration.Companion.seconds
class PacketHandlerImplTest {
@@ -275,4 +282,80 @@ class PacketHandlerImplTest {
verifySuspend { meshLogRepository.insert(any()) }
}
+
+ private fun enrouteDataPacket(id: Int, time: Long = 0L) =
+ DataPacket(to = "!12345678", bytes = null, dataType = 1, id = id, time = time, status = MessageStatus.ENROUTE)
+
+ @Test
+ fun `unacked ENROUTE send times out to a retryable ERROR TIMEOUT`() = runTest(testDispatcher) {
+ connectionStateFlow.value = ConnectionState.Connected
+ everySuspend { packetRepository.getPacketById(123) } returns enrouteDataPacket(123)
+
+ handler.sendToRadio(ToRadio(packet = MeshPacket(id = 123)))
+ testScheduler.advanceTimeBy(PacketHandlerImpl.SEND_ACK_TIMEOUT + 1.seconds)
+ testScheduler.runCurrent()
+
+ verifySuspend { packetRepository.timeOutEnroutePacket(123, Routing.Error.TIMEOUT.value) }
+ }
+
+ @Test
+ fun `the timeout never fires before its deadline`() = runTest(testDispatcher) {
+ connectionStateFlow.value = ConnectionState.Connected
+ everySuspend { packetRepository.getPacketById(124) } returns enrouteDataPacket(124)
+
+ handler.sendToRadio(ToRadio(packet = MeshPacket(id = 124)))
+ testScheduler.advanceTimeBy(PacketHandlerImpl.SEND_ACK_TIMEOUT - 1.seconds)
+ testScheduler.runCurrent()
+
+ verifySuspend(exactly(0)) { packetRepository.timeOutEnroutePacket(any(), any()) }
+ }
+
+ @Test
+ fun `rearm times out a stale persisted ENROUTE packet after the reconnect grace`() = runTest(testDispatcher) {
+ val stale = enrouteDataPacket(321, time = 0L)
+ everySuspend { packetRepository.getEnroutePackets() } returns listOf(stale)
+
+ handler.rearmSendAckTimeouts()
+ testScheduler.advanceTimeBy(PacketHandlerImpl.REARM_GRACE + 1.seconds)
+ testScheduler.runCurrent()
+
+ verifySuspend { packetRepository.timeOutEnroutePacket(321, Routing.Error.TIMEOUT.value) }
+ }
+
+ @Test
+ fun `rearm gives a fresh ENROUTE packet its full ack window`() = runTest(testDispatcher) {
+ val fresh = enrouteDataPacket(322, time = nowMillis)
+ everySuspend { packetRepository.getEnroutePackets() } returns listOf(fresh)
+
+ handler.rearmSendAckTimeouts()
+ testScheduler.advanceTimeBy(PacketHandlerImpl.REARM_GRACE + 1.seconds)
+ testScheduler.runCurrent()
+ verifySuspend(exactly(0)) { packetRepository.timeOutEnroutePacket(any(), any()) }
+
+ testScheduler.advanceTimeBy(PacketHandlerImpl.SEND_ACK_TIMEOUT + 1.seconds)
+ testScheduler.runCurrent()
+ verifySuspend { packetRepository.timeOutEnroutePacket(322, Routing.Error.TIMEOUT.value) }
+ }
+
+ @Test
+ fun `rearming supersedes the pending timer instead of stacking a second one`() = runTest(testDispatcher) {
+ // Repeated reconnects must not accumulate timers for the same send, and the superseded timer must not
+ // fire on its own original deadline.
+ connectionStateFlow.value = ConnectionState.Connected
+ val packet = enrouteDataPacket(325, time = nowMillis)
+ everySuspend { packetRepository.getPacketById(325) } returns packet
+ everySuspend { packetRepository.getEnroutePackets() } returns listOf(packet)
+
+ handler.sendToRadio(ToRadio(packet = MeshPacket(id = 325)))
+ testScheduler.runCurrent()
+ repeat(3) {
+ handler.rearmSendAckTimeouts()
+ testScheduler.runCurrent()
+ }
+
+ testScheduler.advanceTimeBy(PacketHandlerImpl.SEND_ACK_TIMEOUT * 2 + 1.seconds)
+ testScheduler.runCurrent()
+
+ verifySuspend(exactly(1)) { packetRepository.timeOutEnroutePacket(325, Routing.Error.TIMEOUT.value) }
+ }
}
diff --git a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt
index b53c21a808..8570027e37 100644
--- a/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt
+++ b/core/database/src/commonMain/kotlin/org/meshtastic/core/database/dao/PacketDao.kt
@@ -631,6 +631,22 @@ interface PacketDao {
}
}
+ /**
+ * Stamps [routingError] on a sent packet only while it is still [MessageStatus.ENROUTE]. The read and the write
+ * share one transaction so an ACK/NAK that resolves the packet concurrently is never overwritten by a send-ack
+ * timeout that sampled the row before it landed.
+ *
+ * @return true if a row was timed out.
+ */
+ @Transaction
+ suspend fun timeOutEnroutePacket(packetId: Int, routingError: Int): Boolean {
+ val enroute = findPacketsWithId(packetId).filter { it.data.status == MessageStatus.ENROUTE }
+ enroute.forEach { existing ->
+ update(existing.copy(data = existing.data.copy(status = MessageStatus.ERROR), routingError = routingError))
+ }
+ return enroute.isNotEmpty()
+ }
+
/**
* Atomically finds reactions by [replacement]'s packetId + userId + emoji and updates every ownership-scoped copy,
* borrowing [myNodeNum][ReactionEntity.myNodeNum] from each existing row. No-op if no match is found.
diff --git a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonPacketDaoTest.kt b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonPacketDaoTest.kt
index 083c36097e..f7c040e5fc 100644
--- a/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonPacketDaoTest.kt
+++ b/core/database/src/commonTest/kotlin/org/meshtastic/core/database/dao/CommonPacketDaoTest.kt
@@ -29,6 +29,7 @@ import org.meshtastic.core.model.DataPacket
import org.meshtastic.core.model.MessageStatus
import org.meshtastic.core.model.NodeAddress
import org.meshtastic.proto.PortNum
+import org.meshtastic.proto.Routing
import kotlin.test.AfterTest
import kotlin.test.Test
import kotlin.test.assertEquals
@@ -152,6 +153,55 @@ abstract class CommonPacketDaoTest {
assertEquals(MessageStatus.DELIVERED, updatedMessages.first { it.packet.data.id == 999 }.packet.data.status)
}
+ private suspend fun insertSentPacket(packetId: Int, status: MessageStatus) {
+ packetDao.insert(
+ Packet(
+ uuid = 0L,
+ myNodeNum = myNodeNum,
+ port_num = PortNum.TEXT_MESSAGE_APP.value,
+ contact_key = "sent",
+ received_time = nowMillis,
+ read = true,
+ packetId = packetId,
+ data =
+ DataPacket(
+ to = NodeAddress.ID_BROADCAST,
+ bytes = "Sent".encodeToByteArray().toByteString(),
+ dataType = PortNum.TEXT_MESSAGE_APP.value,
+ id = packetId,
+ status = status,
+ ),
+ ),
+ )
+ }
+
+ @Test
+ fun timeOutEnroutePacketFailsOnlyStillEnroutePackets() = runTest {
+ createDb()
+ insertSentPacket(packetId = 8001, status = MessageStatus.ENROUTE)
+
+ assertTrue(packetDao.timeOutEnroutePacket(8001, TIMEOUT_ROUTING_ERROR))
+
+ val timedOut = packetDao.getPacketByPacketId(8001)
+ assertNotNull(timedOut)
+ assertEquals(MessageStatus.ERROR, timedOut.packet.data.status)
+ assertEquals(TIMEOUT_ROUTING_ERROR, timedOut.packet.routingError)
+ }
+
+ @Test
+ fun timeOutEnroutePacketLeavesAnAlreadyResolvedPacketAlone() = runTest {
+ createDb()
+ // The ACK that resolved this packet landed while a send-ack timeout was pending; the timeout must not
+ // overwrite the delivered status it sampled before the ACK arrived.
+ insertSentPacket(packetId = 8002, status = MessageStatus.DELIVERED)
+
+ assertFalse(packetDao.timeOutEnroutePacket(8002, TIMEOUT_ROUTING_ERROR))
+
+ val untouched = packetDao.getPacketByPacketId(8002)
+ assertNotNull(untouched)
+ assertEquals(MessageStatus.DELIVERED, untouched.packet.data.status)
+ }
+
@Test
fun testGetQueuedPackets() = runTest {
createDb()
@@ -323,5 +373,7 @@ abstract class CommonPacketDaoTest {
companion object {
private const val SAMPLE_SIZE = 10
+
+ private val TIMEOUT_ROUTING_ERROR = Routing.Error.TIMEOUT.value
}
}
diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketHandler.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketHandler.kt
index 1ede06c8ee..4a3f364044 100644
--- a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketHandler.kt
+++ b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketHandler.kt
@@ -48,4 +48,11 @@ interface PacketHandler {
/** Stops the packet queue. */
fun stopPacketQueue()
+
+ /**
+ * Re-arms the send-ACK timeout for every persisted packet still awaiting its routing ACK/NAK, so sends orphaned by
+ * a disconnect or app restart become retryable instead of showing as sending forever. Call once per connection,
+ * after the radio config is loaded.
+ */
+ fun rearmSendAckTimeouts()
}
diff --git a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketRepository.kt b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketRepository.kt
index c9c6de0564..458c0725a6 100644
--- a/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketRepository.kt
+++ b/core/repository/src/commonMain/kotlin/org/meshtastic/core/repository/PacketRepository.kt
@@ -73,6 +73,17 @@ interface PacketRepository {
/** Returns all packets currently queued for transmission. */
suspend fun getQueuedPackets(): List<DataPacket>
+ /** Returns all sent packets still awaiting a routing ACK/NAK (status [MessageStatus.ENROUTE]). */
+ suspend fun getEnroutePackets(): List<DataPacket>
+
+ /**
+ * Atomically marks a still-[MessageStatus.ENROUTE] packet as failed with [routingError], leaving it untouched if an
+ * ACK/NAK already resolved it.
+ *
+ * @return true if the packet was timed out.
+ */
+ suspend fun timeOutEnroutePacket(packetId: Int, routingError: Int): Boolean
+
/**
* Persists a packet in the database.
*
Served by rngit 1.4.2 - Generated in 0.15s